{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "8f4ee43e",
   "metadata": {},
   "source": [
    "## FastAPI 对话服务封装方案\n",
    "\n"
   ]
  },
  {
   "cell_type": "raw",
   "id": "fbf67547",
   "metadata": {
    "vscode": {
     "languageId": "raw"
    }
   },
   "source": [
    "chat_service/\n",
    "├── main.py              # FastAPI 主应用\n",
    "├── models.py            # 数据模型定义\n",
    "├── chat_chain.py        # 对话链逻辑\n",
    "├── session_manager.py   # 会话管理\n",
    "└── requirements.txt     # 依赖包"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1b3c7fdb",
   "metadata": {},
   "outputs": [],
   "source": [
    "# main.py\n",
    "from fastapi import FastAPI, HTTPException\n",
    "from pydantic import BaseModel\n",
    "from typing import Dict, List, Optional\n",
    "import uvicorn\n",
    "import asyncio\n",
    "from chat_chain import ChatChain\n",
    "from session_manager import SessionManager\n",
    "\n",
    "app = FastAPI(\n",
    "    title=\"智能对话服务\",\n",
    "    description=\"基于 LangChain 0.3 的对话 API\",\n",
    "    version=\"1.0.0\"\n",
    ")\n",
    "\n",
    "# 全局实例\n",
    "chat_chain = None\n",
    "session_manager = SessionManager()\n",
    "\n",
    "@app.on_event(\"startup\")\n",
    "async def startup_event():\n",
    "    \"\"\"应用启动时初始化\"\"\"\n",
    "    global chat_chain\n",
    "    chat_chain = ChatChain()\n",
    "    await chat_chain.initialize()\n",
    "\n",
    "class ChatRequest(BaseModel):\n",
    "    session_id: str\n",
    "    message: str\n",
    "\n",
    "class ChatResponse(BaseModel):\n",
    "    reply: str\n",
    "    session_id: str\n",
    "\n",
    "@app.post(\"/chat\", response_model=ChatResponse)\n",
    "async def chat_endpoint(request: ChatRequest):\n",
    "    \"\"\"对话接口\"\"\"\n",
    "    try:\n",
    "        # 获取会话历史\n",
    "        history = session_manager.get_history(request.session_id)\n",
    "        \n",
    "        # 调用对话链\n",
    "        reply = await chat_chain.process_message(\n",
    "            message=request.message,\n",
    "            history=history\n",
    "        )\n",
    "        \n",
    "        # 更新会话历史\n",
    "        session_manager.add_message(\n",
    "            session_id=request.session_id,\n",
    "            user_message=request.message,\n",
    "            bot_reply=reply\n",
    "        )\n",
    "        \n",
    "        return ChatResponse(\n",
    "            reply=reply,\n",
    "            session_id=request.session_id\n",
    "        )\n",
    "        \n",
    "    except Exception as e:\n",
    "        raise HTTPException(status_code=500, detail=f\"处理失败: {str(e)}\")\n",
    "\n",
    "@app.get(\"/health\")\n",
    "async def health_check():\n",
    "    \"\"\"健康检查\"\"\"\n",
    "    return {\"status\": \"healthy\", \"langchain_version\": \"0.3.x\"}\n",
    "\n",
    "@app.delete(\"/session/{session_id}\")\n",
    "async def clear_session(session_id: str):\n",
    "    \"\"\"清除会话历史\"\"\"\n",
    "    session_manager.clear_session(session_id)\n",
    "    return {\"message\": f\"会话 {session_id} 已清除\"}\n",
    "\n",
    "@app.get(\"/session/{session_id}/history\")\n",
    "async def get_session_history(session_id: str):\n",
    "    \"\"\"获取会话历史\"\"\"\n",
    "    history = session_manager.get_history(session_id)\n",
    "    return {\"session_id\": session_id, \"history\": history}\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    uvicorn.run(app, host=\"0.0.0.0\", port=8000, reload=True)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b1504148",
   "metadata": {},
   "outputs": [],
   "source": [
    "# chat_chain.py - LangChain 对话链\n",
    "from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder\n",
    "from langchain_core.output_parsers import StrOutputParser\n",
    "from langchain_core.runnables import RunnablePassthrough, RunnableLambda\n",
    "from langchain_core.messages import HumanMessage, AIMessage, SystemMessage\n",
    "from langchain_community.llms import Tongyi\n",
    "from typing import List, Dict, Any\n",
    "import asyncio\n",
    "\n",
    "class ChatChain:\n",
    "    def __init__(self):\n",
    "        self.llm = None\n",
    "        self.chain = None\n",
    "        self.parser = StrOutputParser()\n",
    "        \n",
    "    async def initialize(self):\n",
    "        \"\"\"异步初始化\"\"\"\n",
    "        # 初始化 LLM\n",
    "        self.llm = Tongyi(\n",
    "            temperature=0.7,\n",
    "            model_name=\"qwen-turbo\"  # LangChain 0.3 推荐明确指定模型\n",
    "        )\n",
    "        \n",
    "        # 创建提示模板 - LangChain 0.3 风格\n",
    "        self.prompt = ChatPromptTemplate.from_messages([\n",
    "            (\"system\", \"\"\"你是一个智能客服助手，请遵循以下规则：\n",
    "                1. 友好、专业地回答用户问题\n",
    "                2. 如果不确定答案，诚实地说不知道\n",
    "                3. 保持回答简洁明了\n",
    "                4. 根据对话历史提供连贯的回复\n",
    "                5. 用中文回答\"\"\"),\n",
    "            MessagesPlaceholder(variable_name=\"history\"),\n",
    "            (\"human\", \"{message}\")\n",
    "        ])\n",
    "        \n",
    "        # 构建链 - LangChain 0.3 LCEL 语法\n",
    "        self.chain = (\n",
    "            RunnablePassthrough.assign(\n",
    "                history=RunnableLambda(self._format_history)\n",
    "            )\n",
    "            | self.prompt\n",
    "            | self.llm\n",
    "            | self.parser\n",
    "        )\n",
    "    \n",
    "    async def process_message(self, message: str, history: List[Dict] = None) -> str:\n",
    "        \"\"\"处理用户消息\"\"\"\n",
    "        try:\n",
    "            # 准备输入数据\n",
    "            input_data = {\n",
    "                \"message\": message,\n",
    "                \"raw_history\": history or []\n",
    "            }\n",
    "            \n",
    "            # 异步调用链\n",
    "            response = await self.chain.ainvoke(input_data)\n",
    "            \n",
    "            return response.strip()\n",
    "            \n",
    "        except Exception as e:\n",
    "            print(f\"处理消息时出错: {e}\")\n",
    "            return \"抱歉，我现在无法处理您的请求，请稍后再试。\"\n",
    "    \n",
    "    def _format_history(self, input_data: Dict[str, Any]) -> List:\n",
    "        \"\"\"格式化历史消息为 LangChain 消息格式\"\"\"\n",
    "        history = input_data.get(\"raw_history\", [])\n",
    "        \n",
    "        if not history:\n",
    "            return []\n",
    "        \n",
    "        messages = []\n",
    "        # 只保留最近5轮对话\n",
    "        recent_history = history[-5:] if len(history) > 5 else history\n",
    "        \n",
    "        for item in recent_history:\n",
    "            messages.append(HumanMessage(content=item[\"user_message\"]))\n",
    "            messages.append(AIMessage(content=item[\"bot_reply\"]))\n",
    "        \n",
    "        return messages\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5369b1ae",
   "metadata": {},
   "outputs": [],
   "source": [
    "# session_manager.py - 会话管理器\n",
    "from typing import Dict, List, Optional\n",
    "import time\n",
    "from collections import defaultdict\n",
    "import json\n",
    "from datetime import datetime\n",
    "\n",
    "class SessionManager:\n",
    "    def __init__(self, max_history_length: int = 10):\n",
    "        self.sessions: Dict[str, List[Dict]] = defaultdict(list)\n",
    "        self.max_history_length = max_history_length\n",
    "        self.last_activity: Dict[str, float] = {}\n",
    "    \n",
    "    def get_history(self, session_id: str) -> List[Dict]:\n",
    "        \"\"\"获取会话历史\"\"\"\n",
    "        self._update_activity(session_id)\n",
    "        return self.sessions.get(session_id, [])\n",
    "    \n",
    "    def add_message(self, session_id: str, user_message: str, bot_reply: str):\n",
    "        \"\"\"添加对话记录\"\"\"\n",
    "        self._update_activity(session_id)\n",
    "        \n",
    "        message_record = {\n",
    "            \"user_message\": user_message,\n",
    "            \"bot_reply\": bot_reply,\n",
    "            \"timestamp\": datetime.now().isoformat(),\n",
    "            \"unix_timestamp\": time.time()\n",
    "        }\n",
    "        \n",
    "        self.sessions[session_id].append(message_record)\n",
    "        \n",
    "        # 限制历史长度\n",
    "        if len(self.sessions[session_id]) > self.max_history_length:\n",
    "            self.sessions[session_id] = self.sessions[session_id][-self.max_history_length:]\n",
    "    \n",
    "    def clear_session(self, session_id: str):\n",
    "        \"\"\"清除会话\"\"\"\n",
    "        if session_id in self.sessions:\n",
    "            del self.sessions[session_id]\n",
    "        if session_id in self.last_activity:\n",
    "            del self.last_activity[session_id]\n",
    "    \n",
    "    def get_session_stats(self) -> Dict:\n",
    "        \"\"\"获取会话统计信息\"\"\"\n",
    "        return {\n",
    "            \"total_sessions\": len(self.sessions),\n",
    "            \"active_sessions\": len([\n",
    "                s for s, t in self.last_activity.items()\n",
    "                if time.time() - t < 3600  # 1小时内活跃\n",
    "            ]),\n",
    "            \"total_messages\": sum(len(history) for history in self.sessions.values())\n",
    "        }\n",
    "    \n",
    "    def _update_activity(self, session_id: str):\n",
    "        \"\"\"更新会话活跃时间\"\"\"\n",
    "        self.last_activity[session_id] = time.time()\n",
    "    \n",
    "    def cleanup_inactive_sessions(self, timeout_hours: int = 24):\n",
    "        \"\"\"清理不活跃的会话\"\"\"\n",
    "        current_time = time.time()\n",
    "        timeout_seconds = timeout_hours * 3600\n",
    "        \n",
    "        inactive_sessions = [\n",
    "            session_id for session_id, last_time in self.last_activity.items()\n",
    "            if current_time - last_time > timeout_seconds\n",
    "        ]\n",
    "        \n",
    "        for session_id in inactive_sessions:\n",
    "            self.clear_session(session_id)\n",
    "        \n",
    "        return len(inactive_sessions)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c44a8393",
   "metadata": {},
   "outputs": [],
   "source": [
    "# test_client.py - 测试客户端\n",
    "import requests\n",
    "import json\n",
    "import time\n",
    "\n",
    "class ChatClient:\n",
    "    def __init__(self, base_url: str = \"http://localhost:8000\"):\n",
    "        self.base_url = base_url\n",
    "        self.session_id = f\"test_user_{int(time.time())}\"\n",
    "    \n",
    "    def send_message(self, message: str) -> dict:\n",
    "        \"\"\"发送消息\"\"\"\n",
    "        url = f\"{self.base_url}/chat\"\n",
    "        payload = {\n",
    "            \"session_id\": self.session_id,\n",
    "            \"message\": message\n",
    "        }\n",
    "        \n",
    "        try:\n",
    "            response = requests.post(url, json=payload)\n",
    "            response.raise_for_status()\n",
    "            return response.json()\n",
    "        except requests.exceptions.RequestException as e:\n",
    "            return {\"error\": str(e)}\n",
    "    \n",
    "    def get_history(self) -> dict:\n",
    "        \"\"\"获取历史记录\"\"\"\n",
    "        url = f\"{self.base_url}/session/{self.session_id}/history\"\n",
    "        try:\n",
    "            response = requests.get(url)\n",
    "            response.raise_for_status()\n",
    "            return response.json()\n",
    "        except requests.exceptions.RequestException as e:\n",
    "            return {\"error\": str(e)}\n",
    "    \n",
    "    def clear_session(self) -> dict:\n",
    "        \"\"\"清除会话\"\"\"\n",
    "        url = f\"{self.base_url}/session/{self.session_id}\"\n",
    "        try:\n",
    "            response = requests.delete(url)\n",
    "            response.raise_for_status()\n",
    "            return response.json()\n",
    "        except requests.exceptions.RequestException as e:\n",
    "            return {\"error\": str(e)}\n",
    "\n",
    "def main():\n",
    "    \"\"\"测试主函数\"\"\"\n",
    "    client = ChatClient()\n",
    "    \n",
    "    print(f\"开始测试对话服务 (会话ID: {client.session_id})\")\n",
    "    print(\"=\" * 50)\n",
    "    \n",
    "    # 测试对话\n",
    "    test_messages = [\n",
    "        \"你好\",\n",
    "        \"你能做什么？\",\n",
    "        \"请介绍一下你自己\",\n",
    "        \"谢谢你的帮助\"\n",
    "    ]\n",
    "    \n",
    "    for message in test_messages:\n",
    "        print(f\" 用户: {message}\")\n",
    "        \n",
    "        result = client.send_message(message)\n",
    "        if \"error\" in result:\n",
    "            print(f\" 错误: {result['error']}\")\n",
    "        else:\n",
    "            print(f\" 助手: {result['reply']}\")\n",
    "        \n",
    "        print(\"-\" * 30)\n",
    "        time.sleep(1)\n",
    "    \n",
    "    # 获取历史记录\n",
    "    print(\"\\n 获取对话历史:\")\n",
    "    history = client.get_history()\n",
    "    if \"error\" not in history:\n",
    "        for i, record in enumerate(history.get(\"history\", []), 1):\n",
    "            print(f\"{i}. 用户: {record['user_message']}\")\n",
    "            print(f\"   助手: {record['bot_reply']}\")\n",
    "            print(f\"   时间: {record['timestamp']}\")\n",
    "    \n",
    "    # 清除会话\n",
    "    print(f\"\\n 清除会话: {client.clear_session()}\")\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    main()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3c240638",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 启动服务\n",
    "python main.py\n",
    "\n",
    "# 测试服务\n",
    "python test_client.py\n",
    "\n",
    "# curl 测试\n",
    "!curl -X POST \"http://localhost:8000/chat\" \\\n",
    "     -H \"Content-Type: application/json\" \\\n",
    "     -d '{\"session_id\": \"user_001\", \"message\": \"你好\"}'"
   ]
  },
  {
   "cell_type": "raw",
   "id": "649404b7",
   "metadata": {
    "vscode": {
     "languageId": "raw"
    }
   },
   "source": [
    "# requirements.txt\n",
    "\n",
    "# FastAPI 相关\n",
    "fastapi==0.116.1\n",
    "uvicorn==0.35.0\n",
    "python-multipart==0.0.20\n",
    "\n",
    "# LangChain 0.3 核心包\n",
    "langchain==0.3.27\n",
    "langchain-core==0.3.75\n",
    "langchain-community==0.3.29\n",
    "\n",
    "# 通义千问支持\n",
    "dashscope==1.22.2\n",
    "\n",
    "# 数据处理\n",
    "pydantic==2.11.7\n",
    "pydantic-settings==2.10.1\n",
    "\n",
    "# HTTP 客户端\n",
    "httpx==0.28.1\n",
    "requests==2.32.5\n",
    "\n",
    "# 异步支持\n",
    "aiofiles==24.1.0\n",
    "anyio==4.10.0\n",
    "\n",
    "# 日志和监控\n",
    "python-dotenv==1.1.1\n",
    "\n",
    "# 可选：如果需要其他 LLM 支持\n",
    "# langchain-openai==0.3.32  # OpenAI 支持\n",
    "# langchain-deepseek==0.1.4  # DeepSeek 支持\n",
    "\n",
    "# 开发和测试工具（可选）\n",
    "pytest==8.3.4\n",
    "pytest-asyncio==0.25.0"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "MLOps",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.12.11"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
